
#include "my_cache_layer.h"

////////////////// mem layer //////////////////////
void
my_cache_mem_layer_open(my_cache_mem_layer_t *mem_layer, int nlists,
    int64_t max_item, int64_t max_bytes) {

    max_item = max_item / MY_CACHE_MEM_NUM;
    max_bytes = max_bytes / MY_CACHE_MEM_NUM;
    for (int i = 0; i < MY_CACHE_MEM_NUM; i++) {
        my_slru_cache_open(mem_layer->slru_cache + i, nlists, max_item, max_bytes);
    }
}

void
my_cache_mem_layer_close(my_cache_mem_layer_t *mem_layer) {
    for (int i = 0; i < MY_CACHE_MEM_NUM; i++) {
        my_slru_cache_close(mem_layer->slru_cache + i);
    }
}

my_bool_t
my_cache_mem_layer_get(my_cache_mem_layer_t *mem_layer, my_cache_key_t *key,
    my_buf_chain_t *data) {
    int idx = key->crc32 % MY_CACHE_MEM_NUM;
    return my_slru_cache_get(mem_layer->slru_cache + idx, key, data);
}

void
my_cache_mem_layer_put(my_cache_mem_layer_t *mem_layer, my_cache_key_t *key,
    my_buf_chain_t *data) {
    int idx = key->crc32 % MY_CACHE_MEM_NUM;
    my_slru_cache_put(mem_layer->slru_cache + idx, key, data);
}

void
my_cache_mem_layer_remove(my_cache_mem_layer_t *mem_layer, my_cache_key_t *key) {
    int idx = key->crc32 % MY_CACHE_MEM_NUM;
    my_slru_cache_remove(mem_layer->slru_cache + idx, key);
}

//////////////////////  storage layer ///////////////////

#define MY_CACHE_STORAGE_LAYER_CHASH_REPLACE            500

// the kicked operation will be executed if
// it fails m times within n seconds.
typedef struct {
    my_rbtree_node_t    tree_node;
    my_queue_t          queue;
    int64_t             first_failed_time;
    int64_t             failed_counter;
    my_bool_t           kicked;
    int64_t             key_len;
    my_disk_volume_t    *volume;
} my_cache_storage_layer_item_t;

static inline void
my_cache_storage_layer_item_init(my_cache_storage_layer_item_t *item,
    my_disk_volume_t *volume, my_bool_t kicked) {

    bzero(item, sizeof(my_cache_storage_layer_item_t));
    item->kicked = kicked;
    item->key_len = strlen(volume->name);
    item->volume = volume;
    item->tree_node.key = my_crc32_short(volume->name, item->key_len);
}

static inline int
my_cache_storage_layer_item_cmp(my_cache_storage_layer_item_t *a,
    my_cache_storage_layer_item_t *b) {

    if (a->tree_node.key != b->tree_node.key) {
        return (a->tree_node.key < b->tree_node.key) ? -1 : 1;
    } else if (a->key_len != b->key_len) {
        return (a->key_len < b->key_len) ? -1 : 1;
    }

    return memcmp(a->volume->name, b->volume->name, a->key_len);
}

static void
my_storage_layer_rbtree_insert(my_rbtree_node_t *temp, my_rbtree_node_t *node, 
    my_rbtree_node_t *sentinel) {

    my_rbtree_node_t **p;
    my_cache_storage_layer_item_t *n, *t;

    for (; ;) {
        n = (my_cache_storage_layer_item_t *)node;
        t = (my_cache_storage_layer_item_t *)temp;

        // do not process the same key
        if (my_cache_storage_layer_item_cmp(n, t) < 0) {
            p = &temp->left;
        } else {
            p = &temp->right;
        }

        if (*p == sentinel) {
            break;
        }

        temp = *p;
    }

    *p = node;
    node->parent = temp;
    node->left = sentinel;
    node->right = sentinel;
    my_rbt_red(node);
}

static my_cache_storage_layer_item_t *
my_cache_storage_rbtree_lookup(my_rbtree_t *rbtree, my_disk_volume_t *volume) {
    int rc;
    my_cache_storage_layer_item_t *n;
    my_rbtree_node_t  *node, *sentinel;

    node = rbtree->root;
    sentinel = rbtree->sentinel;

    my_cache_storage_layer_item_t tmp;
    my_cache_storage_layer_item_init(&tmp, volume, MY_FALSE);

    while(node != sentinel) {
        n = (my_cache_storage_layer_item_t *)node;
        rc = my_cache_storage_layer_item_cmp(n, &tmp);

        if (rc < 0) {
            node = node->right;
            continue;
        }

        if (rc > 0) {
            node = node->left;
            continue;
        }

        return n;
    }

    return NULL;
}

void
my_cache_storage_layer_init(my_cache_storage_layer_t *storage_layer) {
    bzero(storage_layer, sizeof(my_cache_storage_layer_t));

    // no need to do this
    // my_spin_lock_init(&storage_layer->lock);
    // my_rwlock_init(&storage_layer->rwlock);

    my_rbtree_init(&storage_layer->tree, &storage_layer->sentinel,
        my_storage_layer_rbtree_insert);

    my_queue_init(&storage_layer->head);
}

void
my_cache_storage_layer_add_storage(my_cache_storage_layer_t *storage_layer,
    my_disk_volume_t *volume, my_bool_t kicked) {

    my_cache_storage_layer_item_t *item =
        my_malloc(sizeof(my_cache_storage_layer_item_t));
    my_cache_storage_layer_item_init(item, volume, kicked);

    my_spin_lock_lock(&storage_layer->lock);
    my_rbtree_insert(&storage_layer->tree, &item->tree_node);
    my_queue_insert_head(&storage_layer->head, &item->queue);
    my_spin_lock_unlock(&storage_layer->lock);
}

static void
my_cache_storage_layer_rebuild_helper(my_cache_storage_layer_t *storage_layer) {
    my_chash_t *new_chash = my_malloc(sizeof(my_chash_t));
    my_chash_init(new_chash, MY_CACHE_STORAGE_LAYER_CHASH_REPLACE);
    my_cache_storage_layer_item_t *item;
    my_queue_t *q = my_queue_last(&storage_layer->head);
    while(q != &storage_layer->head) {
        item = my_queue_data(q, my_cache_storage_layer_item_t, queue);
        my_chash_add_node(new_chash, item->volume->name, item->volume);
        q = q->prev;
    }
    my_chash_build(new_chash);

    my_chash_t *old_chash;
    my_rwlock_wlock(&storage_layer->rwlock);
    old_chash = storage_layer->chash;
    storage_layer->chash = new_chash;
    my_rwlock_wunlock(&storage_layer->rwlock);

    if (old_chash) {
        my_chash_release(old_chash);
    }
}

void
my_cache_storage_layer_build(my_cache_storage_layer_t *storage_layer) {
    my_spin_lock_lock(&storage_layer->lock);
    my_cache_storage_layer_rebuild_helper(storage_layer);
    my_spin_lock_unlock(&storage_layer->lock);
}

static void
my_cache_storage_layer_kicked_helper(my_cache_storage_layer_t *storage_layer,
    my_disk_volume_t *volume) {

    my_cache_storage_layer_item_t *item = 
        my_cache_storage_rbtree_lookup(&storage_layer->tree, volume);
    if (!item) {
        return;
    }

    if (item->kicked) {
        return;
    }

    my_cache_config_kicked_policy_t *kicked_policy = &g_cache_conf.kicked_policy;
    int64_t cur_time_sec = my_nsec_2_sec(get_cache_time());
    if (cur_time_sec - item->first_failed_time >
        kicked_policy->failed_timeout_sec) {
        item->first_failed_time = cur_time_sec;
        item->failed_counter = 0;
    }

    item->failed_counter++;

    if (item->failed_counter <
        kicked_policy->failed_counter_limit) {
        return;
    }

    item->kicked = MY_TRUE;

    my_cache_storage_layer_rebuild_helper(storage_layer);
}

void
my_cache_storage_layer_kicked(my_cache_storage_layer_t *storage_layer,
    my_disk_volume_t *volume) {

    // rebuild chash table my take a long time.
    // so we use trylock to avoid long time blocking.
    // but this may cause the kicked event to be lost.
    // however, kicked event is a frequently repeated event on an active bad disk.
    // so the kicked event lost problem can be ignore.
    if (!my_spin_lock_trylock(&storage_layer->lock)) {
        return;
    }
    my_cache_storage_layer_kicked_helper(storage_layer, volume);
    my_spin_lock_unlock(&storage_layer->lock);
}

my_disk_volume_t *
my_cache_storage_layer_get_volume(my_cache_storage_layer_t *storage_layer, const char *key) {
    my_disk_volume_t *volume;

    my_rwlock_rlock(&storage_layer->rwlock);
    volume = (my_disk_volume_t *)my_chash_get(storage_layer->chash, key);
    my_rwlock_runlock(&storage_layer->rwlock);

    return volume;
}

void
my_cache_storage_layer_release(my_cache_storage_layer_t *storage_layer) {
    my_chash_t *old_chash;
    my_rwlock_wlock(&storage_layer->rwlock);
    old_chash = storage_layer->chash;
    storage_layer->chash = NULL;
    my_rwlock_wunlock(&storage_layer->rwlock);

    if (old_chash) {
        my_chash_release(old_chash);
    }

    my_spin_lock_lock(&storage_layer->lock);
    // my_rbtree_node_t *tree_node;
    // while(!my_rbtree_empty(storage_layer->tree.root, storage_layer->tree.sentinel)) {
    //     tree_node = my_rbtree_min(storage_layer->tree.root, storage_layer->tree.sentinel);
    //     my_rbtree_delete(&chash->tree, tree_node);
    // }
    my_queue_t *q;
    my_cache_storage_layer_item_t *item;
    while(!my_queue_empty(&storage_layer->head)) {
        q = my_queue_last(&storage_layer->head);
        my_queue_remove(q);
        item = my_queue_data(q, my_cache_storage_layer_item_t, queue);
        my_free(item);
    }
    my_spin_lock_unlock(&storage_layer->lock);
}